package com.navinfo.tripanalysis.offline.util;

import com.navinfo.tripanalysis.common.util.BigDecimalUtils;
import org.apache.spark.util.AccumulatorV2;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.math.BigDecimal;


/**
 * 定义Spark累加器更新数值通用方法类
 * @author hmc
 */

public class AccumulatorUtils {

    private static final Logger logger = LoggerFactory.getLogger(AccumulatorUtils.class);



    private static final double EARTH_RADIUS = 6378.137;// 地球半径,单位千米
    public static final double Ea = 6378137;     //   赤道半径,单位米
    public static final double Eb = 6356725;     //   极半径 ,单位米


    private static double rad(double d) {
        //角度转换成弧度
        return d * Math.PI / 180.0;
    }

    /**
     * 计算两个地点的距离
     *
     * @param lat1 第一个纬度,当前点
     * @param lng1 第一个经度
     * @param lat2 第二个纬度,前一点
     * @param lng2 第二个经度
     * @return 两个经纬度的距离
     */
    public static double getDistance(double lat1, double lng1, double lat2, double lng2) {
        double radLat1 = rad(lat1);//纬度
        double radLat2 = rad(lat2);
        double a = radLat1 - radLat2;//两点纬度之差
        double b = rad(lng1) - rad(lng2);//经度之差

        //计算两点之间距离的公式
        double s = 2 * Math.asin(Math.sqrt(
                Math.pow(Math.sin(a / 2), 2) + Math.cos(radLat1) * Math.cos(radLat2) * Math.pow(Math.sin(b / 2), 2)));
        //弧长乘地球半径（半径）
        s = s * EARTH_RADIUS;
        //精确距离的数值（单位千米）
        s = Math.round(s * 100) / 100.0;
        return s;

    }

    /**
     * 计算两个坐标间方向
     * B点在第一象限及Y轴正半轴，Azimuth=A；
     * B在第二象限，Azimuth=360+A；
     * B在第三四象限及Y轴负半轴，Azimuth=180+A。
     * 对于某些系统，再单独设定B位于X正负半轴上的值就可以了，有些系统可以返回arctan（X/0）=90。
     * @param lat1 当前点
     * @param lng1
     * @param lat2 前一点
     * @param lng2
     * @return
     */
    public static double getDirection (double lat1, double lng1, double lat2, double lng2){
        //double direction = Math.atan((lng2-lng1)*Math.cos(lat2));
        double direction;
        if(lat2 == lat1 && lat2 - lat1 ==0){
             direction = 90;
        }else{
             direction = Math.atan(BigDecimalUtils.divide(BigDecimalUtils.multiply(lng2-lng1,Math.cos(lat2),8).doubleValue(),lat2-lat1,8).doubleValue());
             if(lng1 > lng2 && lat1 > lat2){
                 direction = direction + 360;
             }else if(lat1 < lat2){
                 direction = direction + 180;
             }
        }
        return direction;
    }

    /**
     * 通过上一点坐标、里程和方向预估下一个位置点坐标
     * @param LAT
     * @param LON
     * @param distance
     * @param angle
     * @return
     */
    public static Tuple2<Double,Double> getLatLon(double LAT, double LON, double distance, double angle)
    {
        double dx = distance * 1000 * Math.sin(angle * Math.PI / 180.0);
        double dy = distance * 1000 * Math.cos(angle * Math.PI / 180.0);
        double ec = Eb + (Ea-Eb) * (90.0 - LAT) / 90.0;
        double ed = ec * Math.cos(LAT * Math.PI / 180);
        double newLon = (dx / ed + LON * Math.PI / 180.0) * 180.0 / Math.PI;
        double newLat = (dy / ec + LAT * Math.PI / 180.0) * 180.0 / Math.PI;
        return new Tuple2<>(newLat,newLon);
    }

    /**
     * 计算两个坐标差值
     * @param lat1 当前点
     * @param lng1
     * @param lat2 前一点
     * @param lng2
     * @return
     */
    public static double getLatLonDiffence (double lat1, double lng1, double lat2, double lng2){
        double diff = (Math.abs(lat1 - lat2)+ Math.abs(lng1 - lng2))/2;
        return diff;
    }

    /**
     * 从拼接的字符串中提取字段
     * @param str 字符串
     * @param delimiter 分隔符
     * @param field 字段
     * @return 字段值
     */
    public static String getFieldFromConcatString(String str,
                                                  String delimiter, String field) {
        if(str == null || "".equals(str)){return null;}
        String[] fields = str.split(delimiter);
        for(String concatField : fields) {
            String fieldName = concatField.split("=")[0];
            String fieldValue = concatField.split("=")[1];
            if(fieldName.equals(field)) {
                return fieldValue;
            }
        }
        return null;
    }

    /**
     * 从拼接的字符串中给字段设置值
     * @param str 字符串
     * @param delimiter 分隔符
     * @param field 字段名
     * @param newFieldValue 新的field值
     * @return 字段值
     */
    public static String setFieldInConcatString(String str,
                                                String delimiter, String field, String newFieldValue) {
        String[] fields = str.split(delimiter);

        for(int i = 0; i < fields.length; i++) {
            String fieldName = fields[i].split("=")[0];
            if(fieldName.equals(field)) {
                String concatField = fieldName + "=" + newFieldValue;
                fields[i] = concatField;
                break;
            }
        }

        StringBuffer buffer = new StringBuffer();
        for(int i = 0; i < fields.length; i++) {
            buffer.append(fields[i]);
            if(i < fields.length - 1) {
                buffer.append("|");
            }
        }

        return buffer.toString();
    }

    /**
     * 更新传入范围值，并返回更新后的结果
     * @param str 累加器字符串
     * @param delimiter 分隔符
     * @param field 更新范围
     * @param accValue 更新值，默认为1
     * @return
     */
    public static String mergeData(String str,String delimiter,String field ,int accValue){

        String oldValue = getFieldFromConcatString(str,delimiter,field);
        int newValue = Integer.valueOf(oldValue) + accValue;

        return setFieldInConcatString(str,delimiter,field,String.valueOf(newValue));
    }

    /**
     * 不同节点累加器合并
     * @param acc1str 当前累加器value
     * @param delimiter 分隔符
     * @param acc2 被合并累加器
     * @return 合并更新后累加器value
     */
    public static String mergeAccumulator(
            String acc1str, String delimiter, AccumulatorV2<String, String> acc2){

        //取被合并累加器字符串，并按分隔符切分
        String acc2Str = acc2.value();
        String[] acc2Arr = acc2Str.split(delimiter);
        //遍历每个取值范围，分别将每个取值范围的值合并到当前累加器value
        for(String str:acc2Arr){
            //取每一项的名称和累加值
            String field = str.split("=")[0];
            int value = Integer.valueOf(str.split("=")[1]);
            //累加值为0，跳过
            if(value == 0){
                continue;
            }
            //分别将每一项更新到当前累加器
            acc1str = mergeData(acc1str,delimiter,field,value);
        }
        return acc1str;
    }


    public static void main(String[] args) {
        double distance = getDistance(26.235904,116.364897,26.233849,116.371954);
        double direction = getDirection(26.235904,116.364897,26.235904,116.364897);
        Tuple2<Double,Double> latlng = getLatLon(26.235904,116.364897,distance,direction);
        System.out.println("两点间距离:"+distance);
        System.out.println("两点间方向:"+direction);
        System.out.println("预估第二点坐标:x:"+latlng._1+",y:"+latlng._2);
    }

}

